package com.cl.cloud.binlog.handler;

import com.cl.cloud.binlog.content.BinLogApplicationContext;
import com.cl.cloud.binlog.properties.DbConfig;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description:
 * @author: liuzijian
 * @time: 2022-07-08 16:35
 */
@Slf4j
public abstract class AbstractDistributedClientHandler implements DistributedClientHandler {

    /**
     * 客户端线程池
     */
    private ThreadPoolExecutor clientExecutor;

    /**
     * 上下文
     */
    private BinLogApplicationContext context;

    /**
     * 多数据源监听配置
     */
    private Map<String, DbConfig> dbConfig;

    public AbstractDistributedClientHandler(BinLogApplicationContext context) {
        this.context = context;
        this.dbConfig = context.getBinLogProperty().getDbConfig();
        int threadPoolSize = context.getBinLogProperty().getDbConfig().size();
        //只创建核心线程的线程池
        clientExecutor = new ThreadPoolExecutor(threadPoolSize, threadPoolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private final AtomicInteger mThreadNum = new AtomicInteger(1);

            @Override
            public Thread newThread(@Nonnull Runnable runnable) {
                return new Thread(runnable, "BinlogClientExecutorThread:" + mThreadNum.getAndIncrement());
            }
        });
    }

    @Override
    public void start() {
        dbConfig.forEach((key, dbConfig) -> {
            while (!context.isDestory) {
                log.info("开始连接binlog:{}", dbConfig.toString());
                if (!this.requireClient(dbConfig.toString())) {
                    try {
                        log.info("生成客户端并连接binlog");
                        context.getClientFactory().getClient(dbConfig).connect();//这里线程会阻塞
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    } finally {
                        this.releasClient(dbConfig.toString());
                    }
                }
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                BinaryLogClient client = context.getClientFactory().getClient(dbConfig);
                if (client != null && client.isConnected()) {
                    context.getClientFactory().getClient(dbConfig).disconnect();
                }
            } catch (IOException e) {
                log.error("客户端关闭异常", e);
            }
        });
    }

    /**
     * 模板方法：判断是否需要获取客户端（该方法决定是否会重复生成同一个数据库的监听连接 尽量保持一个数据库 一个连接）
     * @param key
     * @return true 需要获取  false 不需要
     */
    abstract boolean requireClient(String key);

    /**
     * 释放客户端
     */
    abstract void releasClient(String key);

    @Override
    public void close() {
        if (!clientExecutor.isShutdown()) {
            clientExecutor.shutdown();
        }
    }
}
